package com.hopu.shop.pro;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.pro.CateParchaseTop10;
import com.hopu.util.HBaseUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.Properties;

/**
 * @Description 统计销量前十的类目
 */
public class CateParchaseTop10Ana {

    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .config("spark.driver.allowMultipleContexts", "true")
                .config("spark.sql.crossJoin.enabled", "true")
                .appName("eraParchaseAna")
                .master("local").getOrCreate();
        JavaSparkContext context = new JavaSparkContext(spark.sparkContext());
        JavaRDD<String> rdd = context.textFile("D://word/user_session.log");
        JavaRDD<CateParchaseTop10> map = rdd.map(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String cid = json.getString("category_id");
            String cname = json.getString("category_name");
            String type = json.getString("event_type");
            return new CateParchaseTop10(cid, cname, type);
        });

        //过滤出购买量
        JavaRDD<CateParchaseTop10> filter = map.filter(t -> "parchase".equals(t.getType()));
        //转换成DataFrame
        Dataset<Row> df = spark.createDataFrame(filter, CateParchaseTop10.class);

        //分组统计
        Dataset<Row> count = df.groupBy("cid", "cname").count();

        //排序
        Dataset<Row> order = count.orderBy(count.col("count").desc());

        //前十
        Dataset<Row> top10 = order.limit(10);

        top10.show();

        //保存到MySQL
        Properties pro = new Properties();
        pro.setProperty("driver", "com.mysql.jdbc.Driver");
        pro.setProperty("user", "root");
        pro.setProperty("password", "123456");
        top10.write().mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.136.200:3306/data_ana?useUnicode=true&characterEncoding=UTF-8", "cate_par_top10", pro);

        //保存到HBase
//        List<Row> rows = top10.collectAsList();
//        String[] columns = {"cid", "cname", "count"};
//        for (Row row: rows) {
//            String[] values = {row.get(0).toString(), row.get(1).toString(), row.get(2).toString()};
//
//            try {
//                HBaseUtils.putOneRowToHbase("cate_parchase_top10", row.getString(0), "info", columns, values);
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        }
    }
}
